import datetime
from datetime import timezone
import urllib
import pandas as pd
from typing import List, Dict, Any, Optional, Tuple, Union

from sqlalchemy import create_engine
from sqlalchemy import text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy import Column, Integer, String, DateTime, Float, BigInteger, SmallInteger
from sqlalchemy.ext.declarative import declarative_base
from pyctp.base.hlog import HLog

# 设置logger
# 设置logger
timestamp = datetime.datetime.now().strftime("%y%m%d_%H%M")
logger = HLog(name=f"SqlHandle_{timestamp}")

Base = declarative_base()


class CodeTable(Base):
    """代码表"""
    __tablename__ = 'CodeTable'
    
    CodeID = Column('CodeID', Integer, primary_key=True, nullable=False, comment='合约代码ID')
    Market = Column('Market', Integer, nullable=False, comment='市场类型')
    MarketDesc = Column('MarketDesc', String(255), nullable=False, comment='市场描述')
    StockCode = Column('StockCode', String(255), nullable=False, comment='交易所原始代码')
    StockName = Column('StockName', String(255), nullable=False, comment='中文简称')
    ExchangeType = Column('ExchangeType', Integer, nullable=False, comment='交易所类型')
    ExchangeTypeDesc = Column('ExchangeTypeDesc', String(255), nullable=False, comment='交易所描述')
    AddDate = Column('AddDate', Integer, nullable=False, comment='添加日期')
    CodeIDStr = Column('CodeIDStr', String(255), nullable=False, comment='代码ID字符串')
    StockCode2 = Column('StockCode2', String(255), nullable=False, comment='备用代码')
    
    __table_args__ = (
        {'schema': 'dbo'},
        # 注意：复合主键在ORM中通过primary_key=True在多个字段上定义
    )


class FutureMinutesHQ(Base):
    """期货分钟行情"""
    __tablename__ = 'FutureMinutesHQ'
    
    ID = Column('ID', Integer, primary_key=True, nullable=False, comment='ID')
    CodeID = Column('CodeID', Integer, nullable=False, comment='合约代码ID')
    dt = Column('dt', Integer, nullable=False, comment='交易日期 yyyyMMdd')
    Time = Column('Time', Integer, nullable=False, comment='时间 HHmm 如0930')
    MinuteValue = Column('MinuteValue', SmallInteger, nullable=False, comment='当日第几分钟 0-1439')
    
    # 价格字段（整型存储，乘以最小跳动）
    hp = Column('hp', Integer, nullable=False, comment='最高价 (×最小跳动)')
    lp = Column('lp', Integer, nullable=False, comment='最低价 (×最小跳动)')
    op = Column('op', Integer, nullable=False, comment='开盘价 (×最小跳动)')
    cp = Column('cp', Integer, nullable=False, comment='收盘价 (×最小跳动)')
    zcp = Column('zcp', Integer, nullable=False, comment='结算价 (×最小跳动)')
    
    # 成交数据
    vol = Column('vol', BigInteger, nullable=False, comment='成交量（手）')
    amt = Column('amt', BigInteger, nullable=False, comment='成交额（元）')
    oi = Column('oi', BigInteger, nullable=False, comment='持仓量（手）')
    
    # 收益率和数据源
    rtn = Column('rtn', Integer, nullable=False, comment='收益率（十万分之一）')
    HQDataSourceType = Column('HQDataSourceType', String(1), nullable=False, comment='数据源类型 1=交易所 2=模拟 3=回测')
    
    # 插入时间
    InsertUTC = Column('InsertUTC', DateTime, nullable=False, default=lambda: datetime.datetime.now(timezone.utc), comment='插入UTC时间')
    
    # 复合主键约束（在SQLAlchemy中通过__table_args__定义）
    __table_args__ = (
        {'schema': 'dbo'},
        # 注意：复合主键在ORM中通过primary_key=True在多个字段上定义
    )


class SQLServerManager:
    """SQL Server数据库管理器 - 使用SQLAlchemy"""
    
    def __init__(self, server: str = 'localhost', database: str = 'master', 
                 user: str = None, password: str = None, trusted_connection: bool = True,
                 driver: str = 'ODBC Driver 17 for SQL Server'):
        """
        初始化数据库管理器
        
        Args:
            server: 服务器地址
            database: 数据库名称
            user: 用户名（Windows认证时可为None）
            password: 密码（Windows认证时可为None）
            trusted_connection: 是否使用Windows认证
            driver: ODBC驱动程序名称
        """
        self.server = server
        self.database = database
        self.user = user
        self.password = password
        self.trusted_connection = trusted_connection
        self.driver = driver
        self.engine = None
        self.SessionLocal = None
        
    def _create_connection_string(self, database_name: str = None) -> str:
        """创建数据库连接字符串"""
        db_name = database_name or self.database
        
        if self.trusted_connection:
            # Windows认证
            connection_string = (
                f"mssql+pyodbc://@{self.server}/{db_name}?"
                f"driver={urllib.parse.quote_plus(self.driver)}&"
                f"trusted_connection=yes"
            )
        else:
            # SQL Server认证
            connection_string = (
                f"mssql+pyodbc://{self.user}:{self.password}@{self.server}/{db_name}?"
                f"driver={urllib.parse.quote_plus(self.driver)}"
            )
        
        return connection_string
    
    def connect(self, database_name: str = None) -> bool:
        """连接到数据库"""
        try:
            connection_string = self._create_connection_string(database_name)
            # 为CREATE DATABASE操作启用autocommit
            self.engine = create_engine(connection_string, echo=False, isolation_level="AUTOCOMMIT")
            
            # 测试连接
            with self.engine.connect() as conn:
                conn.execute(text("SELECT 1"))
            
            # 创建会话工厂
            self.SessionLocal = sessionmaker(bind=self.engine)
            
            logger.info(f"成功连接到数据库: {database_name or self.database}")
            return True
            
        except Exception as e:
            logger.error(f"连接数据库失败: {e}")
            return False
    
    def create_database(self, database_name: str) -> bool:
        """检查数据库是否存在，不存在则创建"""
        try:
            # 连接到master数据库来检查和创建目标数据库
            if not self.connect('master'):
                logger.error("无法连接到master数据库")
                return False
            
            # 检查数据库是否已存在
            with self.engine.connect() as conn:
                result = conn.execute(
                    text(f"SELECT name FROM sys.databases WHERE name = '{database_name}'")
                )
                if result.fetchone():
                    logger.info(f"数据库 {database_name} 已存在，直接使用")
                    return True
                
                # 数据库不存在，创建新数据库
                logger.info(f"数据库 {database_name} 不存在，开始创建...")
                conn.execute(text(f"CREATE DATABASE [{database_name}]"))
                logger.info(f"数据库 {database_name} 创建成功")
            
            return True
            
        except Exception as e:
            logger.error(f"数据库操作失败: {e}")
            return False
    
    def execute_sql_file(self, sql_file_path: str, database_name: str = None) -> bool:
        """
        执行SQL文件
        
        Args:
            sql_file_path: SQL文件路径
            database_name: 可选的数据库名称，如果提供则先切换到该数据库
        """
        try:
            # 如果指定了数据库名称，先切换到该数据库
            if database_name:
                if not self.connect(database_name):
                    logger.error(f"无法连接到数据库: {database_name}")
                    return False
                logger.info(f"已切换到数据库: {database_name}")
            
            if not self.engine:
                logger.error("数据库未连接")
                return False
            
            # 读取SQL文件内容
            with open(sql_file_path, 'r', encoding='utf-8') as file:
                sql_content = file.read()
            
            # 按GO分割SQL语句
            sql_statements = sql_content.split('GO')
            
            with self.engine.connect() as conn:
                executed_count = 0
                failed_count = 0
                
                for i, statement in enumerate(sql_statements):
                    statement = statement.strip()
                    
                    # 跳过空语句
                    if not statement:
                        continue
                        
                    # 跳过纯注释块（以/*开头的多行注释）
                    if statement.startswith('/*') and statement.endswith('*/'):
                        continue
                        
                    # 跳过单行注释
                    if statement.startswith('--'):
                        continue
                        
                    # 跳过USE语句（因为我们已经连接到目标数据库）
                    if statement.upper().startswith('USE '):
                        logger.debug(f"跳过USE语句: {statement[:50]}...")
                        continue
                    
                    # 执行SQL语句
                    try:
                        logger.debug(f"执行SQL语句 {i+1}: {statement[:100]}...")
                        conn.execute(text(statement))
                        executed_count += 1
                    except Exception as stmt_error:
                        # 对于扩展属性错误，只记录警告而不中断执行
                        if "已存在属性" in str(stmt_error) or "MS_Description" in str(stmt_error):
                            logger.warning(f"扩展属性已存在，跳过: {stmt_error}")
                            failed_count += 1
                            continue
                        else:
                            # 其他错误则抛出
                            logger.error(f"SQL语句执行失败: {stmt_error}")
                            logger.error(f"失败的语句: {statement[:200]}...")
                            return False
                
                logger.info(f"SQL文件执行完成: {sql_file_path}，成功执行 {executed_count} 条语句，跳过 {failed_count} 条")
                return True
            
        except Exception as e:
            logger.error(f"执行SQL文件失败: {e}")
            return False
    
    def table_exists(self, table_name: str, schema: str = 'dbo') -> bool:
        """检查表是否存在"""
        try:
            with self.engine.connect() as conn:
                result = conn.execute(text("""
                    SELECT COUNT(*) 
                    FROM INFORMATION_SCHEMA.TABLES 
                    WHERE TABLE_SCHEMA = :schema AND TABLE_NAME = :table_name
                """), {"schema": schema, "table_name": table_name})
                return result.scalar() > 0
        except Exception as e:
            logger.error(f"检查表存在性失败: {e}")
            return False
    
    def create_tables(self, database_name: str, use_sql_files: bool = True) -> bool:
        """创建表结构"""
        try:
            # 连接到目标数据库
            if not self.connect(database_name):
                return False
            
            if use_sql_files:
                # 使用SQL文件创建表
                import os
                current_dir = os.path.dirname(os.path.abspath(__file__))
                
                # 检查并创建CodeTable
                if not self.table_exists('CodeTable'):
                    code_table_sql = os.path.join(current_dir, 'CtreatCodeTable.sql')
                    if os.path.exists(code_table_sql):
                        if not self.execute_sql_file(code_table_sql, database_name):
                            logger.error("CodeTable创建失败")
                            return False
                        logger.info("CodeTable表创建成功")
                    else:
                        logger.warning(f"CodeTable SQL文件不存在: {code_table_sql}")
                else:
                    logger.info("CodeTable表已存在，跳过创建")
                
                # 检查并创建FutureMinutesHQ
                if not self.table_exists('FutureMinutesHQ'):
                    future_hq_sql = os.path.join(current_dir, 'CtreatFutureMinutesHQ.sql')
                    if os.path.exists(future_hq_sql):
                        if not self.execute_sql_file(future_hq_sql, database_name):
                            logger.error("FutureMinutesHQ创建失败")
                            return False
                        logger.info("FutureMinutesHQ表创建成功")
                    else:
                        logger.warning(f"FutureMinutesHQ SQL文件不存在: {future_hq_sql}")
                else:
                    logger.info("FutureMinutesHQ表已存在，跳过创建")
                
                logger.info("表结构检查和创建完成")
            else:
                # 使用SQLAlchemy ORM创建表
                Base.metadata.create_all(bind=self.engine)
                logger.info("使用ORM创建表结构成功")
            
            return True
            
        except Exception as e:
            logger.error(f"创建表结构失败: {e}")
            return False
    
    def get_session(self) -> Session:
        """获取数据库会话"""
        if not self.SessionLocal:
            raise RuntimeError("数据库未连接，请先调用connect方法")
        return self.SessionLocal()
    
    def init_database(self, server: str = 'localhost', database: str = 'FutureDB', 
                     trusted_connection: bool = True) -> bool:
        """
        初始化数据库（创建数据库和表结构）
        
        Args:
            server: 服务器地址，默认'localhost'
            database: 数据库名称，默认'FutureDB'
            trusted_connection: 是否使用Windows认证，默认True
            
        Returns:
            bool: 初始化是否成功
        """
        logger.info("=" * 50)
        logger.info("开始初始化数据库")
        logger.info("=" * 50)
        
        # 更新连接参数
        self.server = server
        self.database = database
        self.trusted_connection = trusted_connection
        
        # 1. 检查并创建数据库
        logger.info(f"1. 检查并创建数据库 '{database}'...")
        if self.create_database(database):
            logger.info("数据库准备完成（已存在或创建成功）")
        else:
            logger.error("数据库准备失败")
            return False
        
        # 2. 创建表结构
        logger.info("2. 创建表结构...")
        if self.create_tables(database, use_sql_files=True):
            logger.info("使用SQL文件创建表结构成功")
        else:
            logger.error("表结构创建失败")
            return False
        
        logger.info(f"数据库 '{database}' 初始化完成")
        return True


class CodeTableManager:
    """代码表管理器"""
    
    def __init__(self, db_manager: SQLServerManager):
        self.db_manager = db_manager
    
    def create(self, CodeID: int, Market: int, MarketDesc: str, StockCode: str, 
               StockName: str, ExchangeType: int, ExchangeTypeDesc: str = None,
               AddDate: int = None, CodeIDStr: str = None, StockCode2: str = None) -> bool:
        """创建代码记录"""
        session = self.db_manager.get_session()
        try:
            code_record = CodeTable(
                CodeID=CodeID,
                Market=Market,
                MarketDesc=MarketDesc,
                StockCode=StockCode,
                StockName=StockName,
                ExchangeType=ExchangeType,
                ExchangeTypeDesc=ExchangeTypeDesc,
                AddDate=AddDate,
                CodeIDStr=CodeIDStr,
                StockCode2=StockCode2
            )
            session.add(code_record)
            session.commit()
            logger.info(f"代码记录创建成功: {StockCode}")
            return True
            
        except SQLAlchemyError as e:
            session.rollback()
            logger.error(f"创建代码记录失败: {e}")
            return False
        finally:
            session.close()
    
    def read(self, StockCode: str = None, Market: int = None, ExchangeType: int = None) -> List[Dict[str, Any]]:
        """
        读取代码记录
        
        Args:
            StockCode: 交易所原始代码，如'CU2511'
            Market: 市场类型，1=期货 2=股票 3=期权
            ExchangeType: 交易所类型，1=CFFEX 2=SHFE 3=DCE 4=CZCE 5=INE 6=GFEX
            
        Returns:
            List[Dict[str, Any]]: 代码记录列表
        """
        session = self.db_manager.get_session()
        try:
            query = session.query(CodeTable)
            
            # 添加过滤条件
            if StockCode:
                query = query.filter(CodeTable.StockCode == StockCode)
            if Market is not None:
                query = query.filter(CodeTable.Market == Market)
            if ExchangeType is not None:
                query = query.filter(CodeTable.ExchangeType == ExchangeType)
            
            records = query.all()
            result = []
            for record in records:
                result.append({
                    'CodeID': record.CodeID,
                    'Market': record.Market,
                    'MarketDesc': record.MarketDesc,
                    'StockCode': record.StockCode,
                    'StockName': record.StockName,
                    'ExchangeType': record.ExchangeType,
                    'ExchangeTypeDesc': record.ExchangeTypeDesc,
                    'AddDate': record.AddDate,
                    'CodeIDStr': record.CodeIDStr,
                    'StockCode2': record.StockCode2
                })
            
            return result
            
        except SQLAlchemyError as e:
            logger.error(f"读取代码记录失败: {e}")
            return []
        finally:
            session.close()
    
    def update(self, CodeID: int, **kwargs) -> bool:
        """更新代码记录"""
        session = self.db_manager.get_session()
        try:
            record = session.query(CodeTable).filter(CodeTable.CodeID == CodeID).first()
            if not record:
                logger.warning(f"未找到代码记录: {CodeID}")
                return False
            
            # 更新字段
            for key, value in kwargs.items():
                if hasattr(record, key):
                    setattr(record, key, value)
            
            session.commit()
            logger.info(f"代码记录更新成功: {CodeID}")
            return True
            
        except SQLAlchemyError as e:
            session.rollback()
            logger.error(f"更新代码记录失败: {e}")
            return False
        finally:
            session.close()
    
    def get_code_id(self, StockCode: str, Market: int = None, ExchangeType: int = None) -> Optional[int]:
        """
        根据StockCode、Market和ExchangeType查询对应的CodeID
        
        Args:
            StockCode: 交易所原始代码，如'CU2511'
            Market: 市场类型，1=期货 2=股票 3=期权（可选，用于精确匹配）
            ExchangeType: 交易所类型，1=CFFEX 2=SHFE 3=DCE 4=CZCE 5=INE 6=GFEX（可选，用于精确匹配）
            
        Returns:
            Optional[int]: 找到的CodeID，如果未找到则返回None
        """
        session = self.db_manager.get_session()
        try:
            query = session.query(CodeTable.CodeID).filter(CodeTable.StockCode == StockCode)
            
            # 添加可选的过滤条件
            if Market is not None:
                query = query.filter(CodeTable.Market == Market)
            if ExchangeType is not None:
                query = query.filter(CodeTable.ExchangeType == ExchangeType)
            
            record = query.first()
            if record:
                return record.CodeID
            else:
                logger.warning(f"未找到匹配的代码记录: StockCode={StockCode}, Market={Market}, ExchangeType={ExchangeType}")
                return None
            
        except SQLAlchemyError as e:
            logger.error(f"查询CodeID失败: {e}")
            return None
        finally:
            session.close()
    
    def delete(self, CodeID: int) -> bool:
        """删除代码记录"""
        session = self.db_manager.get_session()
        try:
            record = session.query(CodeTable).filter(CodeTable.CodeID == CodeID).first()
            if not record:
                logger.warning(f"未找到代码记录: {CodeID}")
                return False
            
            session.delete(record)
            session.commit()
            logger.info(f"代码记录删除成功: {CodeID}")
            return True
            
        except SQLAlchemyError as e:
            session.rollback()
            logger.error(f"删除代码记录失败: {e}")
            return False
        finally:
            session.close()
    
    def import_from_csv(self, csv_file_path: str, batch_size: int = 100) -> Tuple[bool, Dict[str, int]]:
        """
        从CSV文件导入代码表数据
        
        Args:
            csv_file_path: CSV文件路径
            batch_size: 批处理大小，默认100条记录一批
            
        Returns:
            Tuple[bool, Dict[str, int]]: (成功标志, 统计信息)
            统计信息包含: {'total': 总数, 'inserted': 插入数, 'updated': 更新数, 'errors': 错误数}
        """
        try:
            # 读取CSV文件
            df = pd.read_csv(csv_file_path, encoding='utf-8')
            logger.info(f"成功读取CSV文件: {csv_file_path}, 共{len(df)}条记录")
            
            # 检查必要的列是否存在
            required_columns = ['CodeID', 'Market', 'MarketDesc', 'stock_code', 'stock_name', 'ExchangeType']
            missing_columns = [col for col in required_columns if col not in df.columns]
            if missing_columns:
                logger.error(f"CSV文件缺少必要的列: {missing_columns}")
                return False, {'total': 0, 'inserted': 0, 'updated': 0, 'errors': len(df)}
            
            # 统计信息
            stats = {'total': len(df), 'inserted': 0, 'updated': 0, 'errors': 0}
            
            # 批量处理数据
            for i in range(0, len(df), batch_size):
                batch_df = df.iloc[i:i+batch_size]
                logger.info(f"处理批次 {i//batch_size + 1}: 记录 {i+1}-{min(i+batch_size, len(df))}")
                
                for _, row in batch_df.iterrows():
                    try:
                        # 准备数据，处理NaN值
                        code_data = {
                            'CodeID': int(row['CodeID']),
                            'Market': int(row['Market']),
                            'MarketDesc': str(row['MarketDesc']),
                            'StockCode': str(row['stock_code']),  # CSV中是stock_code
                            'StockName': str(row['stock_name']),  # CSV中是stock_name
                            'ExchangeType': int(row['ExchangeType']),
                            'ExchangeTypeDesc': str(row['ExchangeTypeDesc']) if pd.notna(row['ExchangeTypeDesc']) else None,
                            'AddDate': int(row['AddDate']) if pd.notna(row['AddDate']) and row['AddDate'] != 0 else None,
                            'CodeIDStr': str(row['CodeIDStr']) if pd.notna(row['CodeIDStr']) else None,
                            'StockCode2': str(row['stock_code2']) if pd.notna(row['stock_code2']) else None  # CSV中是stock_code2
                        }
                        
                        # 调用create_or_update方法
                        success, operation = self.create_or_update(**code_data)
                        
                        if success:
                            if operation == "INSERT":
                                stats['inserted'] += 1
                            elif operation == "UPDATE":
                                stats['updated'] += 1
                        else:
                            stats['errors'] += 1
                            logger.error(f"处理记录失败: CodeID={code_data['CodeID']}, StockCode={code_data['StockCode']}")
                            
                    except Exception as e:
                        stats['errors'] += 1
                        logger.error(f"处理记录时发生异常: {e}, 行数据: {row.to_dict()}")
            
            # 输出统计信息
            logger.info(f"CSV导入完成 - 总计: {stats['total']}, 插入: {stats['inserted']}, 更新: {stats['updated']}, 错误: {stats['errors']}")
            
            return stats['errors'] == 0, stats
            
        except FileNotFoundError:
            logger.error(f"CSV文件不存在: {csv_file_path}")
            return False, {'total': 0, 'inserted': 0, 'updated': 0, 'errors': 1}
        except pd.errors.EmptyDataError:
            logger.error(f"CSV文件为空: {csv_file_path}")
            return False, {'total': 0, 'inserted': 0, 'updated': 0, 'errors': 1}
        except Exception as e:
            logger.error(f"导入CSV文件时发生异常: {e}")
            return False, {'total': 0, 'inserted': 0, 'updated': 0, 'errors': 1}

    def create_or_update(self, CodeID: int, Market: int, MarketDesc: str, StockCode: str, 
                        StockName: str, ExchangeType: int, ExchangeTypeDesc: str = None,
                        AddDate: int = None, CodeIDStr: str = None, StockCode2: str = None) -> Tuple[bool, str]:
        """
        插入或更新代码记录
        
        Args:
            CodeID: 合约代码ID（主键）
            Market: 市场类型
            MarketDesc: 市场描述
            StockCode: 交易所原始代码
            StockName: 中文简称
            ExchangeType: 交易所类型
            ExchangeTypeDesc: 交易所描述
            AddDate: 添加日期
            CodeIDStr: 代码ID字符串
            StockCode2: 备用代码
            
        Returns:
            Tuple[bool, str]: (成功标志, 操作类型) - 操作类型为 "INSERT" 或 "UPDATE"
        """
        session = self.db_manager.get_session()
        try:
            # 根据 Market 和 StockCode 查询是否存在相同记录
            existing_record = session.query(CodeTable).filter(
                CodeTable.Market == Market,
                CodeTable.StockCode == StockCode
            ).first()
            
            if existing_record:
                # 更新现有记录
                #existing_record.CodeID = CodeID  # 更新CodeID
                existing_record.MarketDesc = MarketDesc
                existing_record.StockName = StockName
                existing_record.ExchangeType = ExchangeType
                existing_record.ExchangeTypeDesc = ExchangeTypeDesc
                existing_record.AddDate = AddDate
                existing_record.CodeIDStr = CodeIDStr
                existing_record.StockCode2 = StockCode2
                
                session.commit()
                logger.info(f"代码记录更新成功: {StockCode} (保持原有CodeID: {existing_record.CodeID})")
                return True, "UPDATE"
            else:
                # 插入新记录
                new_record = CodeTable(
                    CodeID=CodeID,
                    Market=Market,
                    MarketDesc=MarketDesc,
                    StockCode=StockCode,
                    StockName=StockName,
                    ExchangeType=ExchangeType,
                    ExchangeTypeDesc=ExchangeTypeDesc,
                    AddDate=AddDate,
                    CodeIDStr=CodeIDStr,
                    StockCode2=StockCode2
                )
                session.add(new_record)
                session.commit()
                logger.info(f"代码记录插入成功: {StockCode} (CodeID: {CodeID})")
                return True, "INSERT"
                
        except SQLAlchemyError as e:
            session.rollback()
            logger.error(f"插入或更新代码记录失败: {e}")
            return False, "ERROR"
        finally:
            session.close()
    
    def get_max_code_id(self) -> Optional[int]:
        """
        获取CodeTable表中最大的CodeID值
        
        Returns:
            Optional[int]: 最大的CodeID值，如果表为空则返回None
        """
        session = self.db_manager.get_session()
        try:
            from sqlalchemy import func
            result = session.query(func.max(CodeTable.CodeID)).scalar()
            
            if result is not None:
                logger.info(f"当前最大CodeID: {result}")
                return result
            else:
                logger.info("CodeTable表为空，没有找到CodeID")
                return None
                
        except SQLAlchemyError as e:
            logger.error(f"查询最大CodeID失败: {e}")
            return None
        finally:
            session.close()


class FutureMinutesHQManager:
    """期货分钟行情管理器"""
    
    def __init__(self, db_manager: SQLServerManager):
        self.db_manager = db_manager
    
    def create(self, **kwargs) -> Tuple[bool, str, str]:
        """
        创建行情记录
        
        Returns:
            Tuple[bool, str, str]: (成功标志, 错误码, 错误信息)
            - 成功时返回: (True, "SUCCESS", "")
            - 失败时返回: (False, 错误码, 错误信息)
        """
        session = self.db_manager.get_session()
        try:
            hq_record = FutureMinutesHQ(**kwargs)
            session.add(hq_record)
            session.commit()
            logger.info(f"行情记录创建成功: {kwargs.get('CodeID')}")
            return True, "SUCCESS", ""
            
        except IntegrityError as e:
            session.rollback()
            error_msg = str(e)
            if "PRIMARY KEY" in error_msg or "PK_" in error_msg:
                error_code = "DUPLICATE_KEY"
                error_info = f"主键冲突: CodeID={kwargs.get('CodeID')}, dt={kwargs.get('dt')}, Time={kwargs.get('Time')}"
            elif "UNIQUE" in error_msg:
                error_code = "UNIQUE_CONSTRAINT"
                error_info = "唯一约束冲突"
            else:
                error_code = "INTEGRITY_ERROR"
                error_info = "数据完整性错误"
            
            logger.error(f"创建行情记录失败 [{error_code}]: {error_info}")
            return False, error_code, error_info
            
        except SQLAlchemyError as e:
            session.rollback()
            error_code = "SQL_ERROR"
            error_info = str(e)
            logger.error(f"创建行情记录失败 [{error_code}]: {error_info}")
            return False, error_code, error_info
        finally:
            session.close()
    
    def read(self, CodeID: int = None, dt: int = None, 
             limit: int = 100) -> List[Dict[str, Any]]:
        """读取行情记录"""
        session = self.db_manager.get_session()
        try:
            query = session.query(FutureMinutesHQ)
            
            if CodeID:
                query = query.filter(FutureMinutesHQ.CodeID == CodeID)
            if dt:
                query = query.filter(FutureMinutesHQ.dt == dt)
            
            query = query.order_by(FutureMinutesHQ.InsertUTC.desc()).limit(limit)
            records = query.all()
            
            result = []
            for record in records:
                result.append({
                    'ID': record.ID,
                    'CodeID': record.CodeID,
                    'dt': record.dt,
                    'Time': record.Time,
                    'MinuteValue': record.MinuteValue,
                    'hp': record.hp,
                    'lp': record.lp,
                    'op': record.op,
                    'cp': record.cp,
                    'zcp': record.zcp,
                    'vol': record.vol,
                    'amt': record.amt,
                    'oi': record.oi,
                    'rtn': record.rtn,
                    'HQDataSourceType': record.HQDataSourceType,
                    'InsertUTC': record.InsertUTC
                })
            
            return result
            
        except SQLAlchemyError as e:
            logger.error(f"读取行情记录失败: {e}")
            return []
        finally:
            session.close()
    
    def update(self, record_id: int, **kwargs) -> bool:
        """更新行情记录"""
        session = self.db_manager.get_session()
        try:
            record = session.query(FutureMinutesHQ).filter(FutureMinutesHQ.ID == record_id).first()
            if not record:
                logger.warning(f"未找到行情记录: {record_id}")
                return False
            
            # 更新字段
            for key, value in kwargs.items():
                if hasattr(record, key):
                    setattr(record, key, value)
            
            session.commit()
            logger.info(f"行情记录更新成功: {record_id}")
            return True
            
        except SQLAlchemyError as e:
            session.rollback()
            logger.error(f"更新行情记录失败: {e}")
            return False
        finally:
            session.close()
    
    def delete(self, record_id: int) -> bool:
        """删除行情记录"""
        session = self.db_manager.get_session()
        try:
            record = session.query(FutureMinutesHQ).filter(FutureMinutesHQ.ID == record_id).first()
            if not record:
                logger.warning(f"未找到行情记录: {record_id}")
                return False
            
            session.delete(record)
            session.commit()
            logger.info(f"行情记录删除成功: {record_id}")
            return True
            
        except SQLAlchemyError as e:
            session.rollback()
            logger.error(f"删除行情记录失败: {e}")
            return False
        finally:
            session.close()
    
    def batch_create(self, records: List[Dict[str, Any]]) -> Tuple[bool, str, str, List[Dict[str, Any]]]:
        """
        批量创建行情记录
        
        Returns:
            Tuple[bool, str, str, List[Dict]]: (成功标志, 错误码, 错误信息, 失败记录列表)
            - 成功时返回: (True, "SUCCESS", "", [])
            - 失败时返回: (False, 错误码, 错误信息, 失败的记录列表)
        """
        session = self.db_manager.get_session()
        try:
            hq_records = [FutureMinutesHQ(**record) for record in records]
            session.add_all(hq_records)
            session.commit()
            logger.info(f"批量创建行情记录成功，共 {len(records)} 条")
            return True, "SUCCESS", "", []
            
        except IntegrityError as e:
            session.rollback()
            error_msg = str(e)
            
            # 解析重复键信息
            failed_records = []
            if "PRIMARY KEY" in error_msg or "PK_" in error_msg:
                error_code = "DUPLICATE_KEY"
                
                # 尝试解析重复键值
                import re
                duplicate_match = re.search(r'重复键值为 \(([^)]+)\)', error_msg)
                if duplicate_match:
                    duplicate_values = duplicate_match.group(1).split(', ')
                    if len(duplicate_values) >= 3:
                        try:
                            code_id = int(duplicate_values[0])
                            dt = int(duplicate_values[1])
                            time = int(duplicate_values[2])
                            
                            # 找出冲突的记录
                            for record in records:
                                if (record.get('CodeID') == code_id and 
                                    record.get('dt') == dt and 
                                    record.get('Time') == time):
                                    failed_records.append(record)
                        except (ValueError, IndexError):
                            pass
                
                error_info = f"主键冲突，重复键值: {duplicate_match.group(1) if duplicate_match else '未知'}"
            elif "UNIQUE" in error_msg:
                error_code = "UNIQUE_CONSTRAINT"
                error_info = "唯一约束冲突"
            else:
                error_code = "INTEGRITY_ERROR"
                error_info = "数据完整性错误"
            
            logger.error(f"批量创建行情记录失败 [{error_code}]: {error_info}")
            return False, error_code, error_info, failed_records
            
        except SQLAlchemyError as e:
            session.rollback()
            error_code = "SQL_ERROR"
            error_info = str(e)
            logger.error(f"批量创建行情记录失败 [{error_code}]: {error_info}")
            return False, error_code, error_info, []
        finally:
            session.close()
    
    def query_klines(self, 
                     code_ids: Union[int, List[int]], 
                     start_date: int = None, 
                     end_date: int = None,
                     start_time: int = None,
                     end_time: int = None) -> pd.DataFrame:
        """
        查询K线数据
        
        Args:
            code_ids: 合约代码ID，可以是单个ID或ID列表
            start_date: 开始日期 yyyyMMdd格式，如20250101
            end_date: 结束日期 yyyyMMdd格式，如20251231
            start_time: 开始时间 HHmm格式，如0930
            end_time: 结束时间 HHmm格式，如1500
            
        Returns:
            pandas.DataFrame: K线数据
        """
        session = self.db_manager.get_session()
        try:
            query = session.query(FutureMinutesHQ)
            
            # 处理CodeID参数
            if isinstance(code_ids, int):
                query = query.filter(FutureMinutesHQ.CodeID == code_ids)
            elif isinstance(code_ids, list):
                query = query.filter(FutureMinutesHQ.CodeID.in_(code_ids))
            
            # 处理日期范围
            if start_date:
                query = query.filter(FutureMinutesHQ.dt >= start_date)
            if end_date:
                query = query.filter(FutureMinutesHQ.dt <= end_date)
            
            # 处理时间范围
            if start_time:
                query = query.filter(FutureMinutesHQ.Time >= start_time)
            if end_time:
                query = query.filter(FutureMinutesHQ.Time <= end_time)
            
            # 按日期和时间排序
            query = query.order_by(FutureMinutesHQ.dt.asc(), FutureMinutesHQ.Time.asc())
            
            records = query.all()
            
            # 转换为DataFrame
            if not records:
                # 返回空DataFrame但包含所有列
                columns = ['ID', 'CodeID', 'dt', 'Time', 'MinuteValue', 'hp', 'lp', 'op', 'cp', 'zcp', 
                          'vol', 'amt', 'oi', 'rtn', 'HQDataSourceType', 'InsertUTC']
                return pd.DataFrame(columns=columns)
            
            data = []
            for record in records:
                data.append({
                    'ID': record.ID,
                    'CodeID': record.CodeID,
                    'dt': record.dt,
                    'Time': record.Time,
                    'MinuteValue': record.MinuteValue,
                    'hp': record.hp,
                    'lp': record.lp,
                    'op': record.op,
                    'cp': record.cp,
                    'zcp': record.zcp,
                    'vol': record.vol,
                    'amt': record.amt,
                    'oi': record.oi,
                    'rtn': record.rtn,
                    'HQDataSourceType': record.HQDataSourceType,
                    'InsertUTC': record.InsertUTC
                })
            
            df = pd.DataFrame(data)
            logger.info(f"查询K线数据成功，共 {len(df)} 条记录")
            return df
            
        except SQLAlchemyError as e:
            logger.error(f"查询K线数据失败: {e}")
            return pd.DataFrame()
        finally:
            session.close()
    
    def create_or_update(self, CodeID: int, dt: int, Time: int, **kwargs) -> Tuple[bool, str, str]:
        """
        插入或更新行情记录
        
        Args:
            CodeID: 合约代码ID
            dt: 交易日期 yyyyMMdd
            Time: 时间 HHmm
            **kwargs: 其他行情字段
            
        Returns:
            Tuple[bool, str, str]: (成功标志, 操作类型, 错误信息)
            - 成功时返回: (True, "INSERT"/"UPDATE", "")
            - 失败时返回: (False, "ERROR", 错误信息)
        """
        session = self.db_manager.get_session()
        try:
            # 检查记录是否已存在（基于复合主键：CodeID, dt, Time）
            existing_record = session.query(FutureMinutesHQ).filter(
                FutureMinutesHQ.CodeID == CodeID,
                FutureMinutesHQ.dt == dt,
                FutureMinutesHQ.Time == Time
            ).first()
            
            if existing_record:
                # 更新现有记录
                for key, value in kwargs.items():
                    if hasattr(existing_record, key) and key not in ['CodeID', 'dt', 'Time']:  # 不更新主键字段
                        setattr(existing_record, key, value)
                
                # 更新插入时间
                existing_record.InsertUTC = datetime.datetime.now(timezone.utc)
                
                session.commit()
                logger.info(f"行情记录更新成功: CodeID={CodeID}, dt={dt}, Time={Time}")
                return True, "UPDATE", ""
            else:
                # 插入新记录
                record_data = {
                    'CodeID': CodeID,
                    'dt': dt,
                    'Time': Time,
                    **kwargs
                }
                
                # 确保必需字段有默认值
                if 'MinuteValue' not in record_data:
                    # 计算当日第几分钟
                    hour = Time // 100
                    minute = Time % 100
                    record_data['MinuteValue'] = hour * 60 + minute
                
                if 'HQDataSourceType' not in record_data:
                    record_data['HQDataSourceType'] = '1'  # 默认为交易所数据
                
                # 确保必需的价格字段存在
                required_fields = ['hp', 'lp', 'op', 'cp', 'zcp', 'vol', 'amt', 'oi', 'rtn']
                for field in required_fields:
                    if field not in record_data:
                        record_data[field] = 0  # 设置默认值
                
                new_record = FutureMinutesHQ(**record_data)
                session.add(new_record)
                session.flush()  # 立即执行以检测约束冲突
                session.commit()
                logger.info(f"行情记录插入成功: CodeID={CodeID}, dt={dt}, Time={Time}")
                return True, "INSERT", ""
                
        except IntegrityError as e:
            session.rollback()
            error_msg = str(e)
            # 如果是主键冲突，尝试再次查询并更新
            if "PRIMARY KEY" in error_msg or "UNIQUE" in error_msg:
                try:
                    existing_record = session.query(FutureMinutesHQ).filter(
                        FutureMinutesHQ.CodeID == CodeID,
                        FutureMinutesHQ.dt == dt,
                        FutureMinutesHQ.Time == Time
                    ).first()
                    
                    if existing_record:
                        # 更新现有记录
                        for key, value in kwargs.items():
                            if hasattr(existing_record, key) and key not in ['CodeID', 'dt', 'Time']:
                                setattr(existing_record, key, value)
                        
                        existing_record.InsertUTC = datetime.datetime.now(timezone.utc)
                        session.commit()
                        logger.info(f"行情记录更新成功（重试）: CodeID={CodeID}, dt={dt}, Time={Time}")
                        return True, "UPDATE", ""
                except Exception as retry_e:
                    logger.error(f"重试更新失败: {retry_e}")
            
            logger.error(f"插入或更新行情记录失败（完整性约束）: {error_msg}")
            return False, "ERROR", error_msg
        except SQLAlchemyError as e:
            session.rollback()
            error_msg = str(e)
            logger.error(f"插入或更新行情记录失败: {error_msg}")
            return False, "ERROR", error_msg
        finally:
            session.close()
    
    def batch_create_or_update(self, records: List[Dict[str, Any]]) -> Tuple[bool, str, str, Dict[str, int]]:
        """
        批量插入或更新行情记录
        
        Args:
            records: 行情记录列表，每个记录必须包含 CodeID, dt, Time 字段
            
        Returns:
            Tuple[bool, str, str, Dict[str, int]]: (成功标志, 错误码, 错误信息, 统计信息)
            - 成功时返回: (True, "SUCCESS", "", {"inserted": 插入数量, "updated": 更新数量})
            - 失败时返回: (False, 错误码, 错误信息, {"inserted": 0, "updated": 0})
        """
        if not records:
            return True, "SUCCESS", "", {"inserted": 0, "updated": 0}
        
        session = self.db_manager.get_session()
        try:
            inserted_count = 0
            updated_count = 0
            
            for record in records:
                # 验证必需字段
                if not all(key in record for key in ['CodeID', 'dt', 'Time']):
                    raise ValueError("每条记录必须包含 CodeID, dt, Time 字段")
                
                CodeID = record['CodeID']
                dt = record['dt']
                Time = record['Time']
                
                # 检查记录是否已存在
                existing_record = session.query(FutureMinutesHQ).filter(
                    FutureMinutesHQ.CodeID == CodeID,
                    FutureMinutesHQ.dt == dt,
                    FutureMinutesHQ.Time == Time
                ).first()
                
                if existing_record:
                    # 更新现有记录（排除主键字段）
                    primary_key_fields = {'CodeID', 'dt', 'Time'}
                    for key, value in record.items():
                        if key not in primary_key_fields and hasattr(existing_record, key):
                            setattr(existing_record, key, value)
                    
                    # 更新插入时间
                    existing_record.InsertUTC = datetime.datetime.now(timezone.utc)
                    updated_count += 1
                else: 
                    # 插入新记录
                    record_data = record.copy()
                    
                    # 确保必需字段有默认值
                    if 'MinuteValue' not in record_data:
                        # 计算当日第几分钟
                        hour = Time // 100
                        minute = Time % 100
                        record_data['MinuteValue'] = hour * 60 + minute
                    
                    if 'HQDataSourceType' not in record_data:
                        record_data['HQDataSourceType'] = '1'  # 默认为交易所数据
                    
                    # 确保所有必需的价格字段有默认值
                    price_fields = ['hp', 'lp', 'op', 'cp', 'zcp', 'vol', 'amt', 'oi', 'rtn']
                    for field in price_fields:
                        if field not in record_data:
                            record_data[field] = 0
                    
                    # 设置插入时间
                    if 'InsertUTC' not in record_data:
                        record_data['InsertUTC'] = datetime.datetime.now(timezone.utc)
                    
                    try:
                        new_record = FutureMinutesHQ(**record_data)
                        session.add(new_record)
                        session.flush()  # 立即执行以检测约束冲突
                        inserted_count += 1
                    except IntegrityError as ie:
                        session.rollback()
                        logger.warning(f"插入记录时遇到约束冲突，尝试更新: CodeID={CodeID}, dt={dt}, Time={Time}")
                        
                        # 重新查询可能存在的记录（可能是并发插入导致的）
                        existing_record = session.query(FutureMinutesHQ).filter(
                            FutureMinutesHQ.CodeID == CodeID,
                            FutureMinutesHQ.dt == dt,
                            FutureMinutesHQ.Time == Time
                        ).first()
                        
                        if existing_record:
                            # 更新现有记录（排除主键字段）
                            primary_key_fields = {'CodeID', 'dt', 'Time'}
                            for key, value in record.items():
                                if key not in primary_key_fields and hasattr(existing_record, key):
                                    setattr(existing_record, key, value)
                            
                            existing_record.InsertUTC = datetime.datetime.now(timezone.utc)
                            updated_count += 1
                            logger.info(f"成功更新记录: CodeID={CodeID}, dt={dt}, Time={Time}")
                        else:
                            # 如果仍然找不到记录，重新抛出异常
                            raise ie
            
            session.commit()
            stats = {"inserted": inserted_count, "updated": updated_count}
            logger.info(f"批量插入或更新行情记录成功: 插入 {inserted_count} 条，更新 {updated_count} 条")
            return True, "SUCCESS", "", stats
            
        except ValueError as e:
            session.rollback()
            error_msg = str(e)
            logger.error(f"批量插入或更新行情记录失败 [VALIDATION_ERROR]: {error_msg}")
            return False, "VALIDATION_ERROR", error_msg, {"inserted": 0, "updated": 0}
            
        except IntegrityError as e:
            session.rollback()
            error_msg = f"数据完整性约束冲突: {str(e)}"
            logger.error(f"批量插入或更新行情记录失败 [INTEGRITY_ERROR]: {error_msg}")
            return False, "INTEGRITY_ERROR", error_msg, {"inserted": 0, "updated": 0}
            
        except SQLAlchemyError as e:
            session.rollback()
            error_msg = str(e)
            logger.error(f"批量插入或更新行情记录失败 [SQL_ERROR]: {error_msg}")
            return False, "SQL_ERROR", error_msg, {"inserted": 0, "updated": 0}
        finally:
            session.close()


def test_database_operations():
    """测试数据库操作功能"""
    logger.info("=" * 50)
    logger.info("开始测试数据库操作功能 - SQLAlchemy版本")
    logger.info("=" * 50)
    
    # 初始化数据库管理器（使用Windows认证）
    db_manager = SQLServerManager(
        server='localhost',
        database='FutureDB',
        trusted_connection=True
    )
    
    # 使用新的init_database方法初始化数据库
    if not db_manager.init_database(
        server='localhost',
        database='FutureDB',
        trusted_connection=True
    ):
        logger.error("数据库初始化失败")
        return
    
    # 3. 测试代码表操作
    logger.info("\n3. 测试代码表操作...")
    code_manager = CodeTableManager(db_manager)
    
    max_code_id = code_manager.get_max_code_id()
    if max_code_id is None:
        logger.war("未找到最大CodeID")
        return


    # 创建测试数据
    test_codes = [
        {'CodeID': 1, 'Market': 1, 'MarketDesc': '期货', 'StockCode': 'CU2511', 'StockName': '沪铜2511', 'ExchangeType': 2, 'ExchangeTypeDesc': 'SHFE'},
        {'CodeID': 2, 'Market': 1, 'MarketDesc': '期货', 'StockCode': 'AU2512', 'StockName': '沪金2512', 'ExchangeType': 2, 'ExchangeTypeDesc': 'SHFE'},
        {'CodeID': 3, 'Market': 1, 'MarketDesc': '期货', 'StockCode': 'IF2510', 'StockName': '沪深300股指2510', 'ExchangeType': 1, 'ExchangeTypeDesc': 'CFFEX'}
    ]
    
    for code_data in test_codes:
        if code_manager.create(**code_data):
            logger.info(f"创建代码记录: {code_data['StockCode']}")
        else:
            logger.error(f"创建代码记录失败: {code_data['StockCode']}")
    
    # 读取数据
    logger.info("读取所有代码记录:")
    records = code_manager.read()
    for record in records:
        logger.info(f"    {record['StockCode']} - {record['StockName']} ({record['ExchangeTypeDesc']})")
    
    # 更新数据
    if code_manager.update(1, StockName='沪铜2511合约'):
        logger.info("更新代码记录成功")
    
    # 4. 测试期货行情表操作
    logger.info("4. 测试期货行情表操作...")
    hq_manager = FutureMinutesHQManager(db_manager)
    
    # 创建测试行情数据

    codeid = code_manager.get_code_id('CU2511', Market=1, ExchangeType=2)
    if codeid is None:
        logger.error("未找到CU2511的CodeID")
        return
    
    test_hq_data = {
        'CodeID': codeid,  # 对应CU2511的CodeID
        'dt': 20250929,  # 交易日期 yyyyMMdd
        'Time': 1430,  # 时间 HHmm
        'MinuteValue': 870,  # 当日第几分钟 (14:30 = 14*60+30 = 870)
        'hp': 377100,  # 最高价 (75420 * 5 = 377100，假设最小跳动为5)
        'lp': 376000,  # 最低价 (75200 * 5)
        'op': 376500,  # 开盘价 (75300 * 5)
        'cp': 377100,  # 收盘价 (75420 * 5)
        'zcp': 377100,  # 结算价 (75420 * 5)
        'vol': 12345,  # 成交量（手）
        'amt': 928765432,  # 成交额（元）
        'oi': 54321,  # 持仓量（手）
        'rtn': 50,  # 收益率（十万分之一）
        'HQDataSourceType': '1'  # 数据源类型 1=交易所
    }
    
    success, error_code, error_msg = hq_manager.create(**test_hq_data)
    if success:
        logger.info("创建行情记录成功")
    else:
        logger.error(f"创建行情记录失败 [{error_code}]: {error_msg}")
    
    # 读取行情数据
    logger.info("读取行情记录:")
    hq_records = hq_manager.read(CodeID=1)
    for record in hq_records:
        logger.info(f"    CodeID: {record['CodeID']} - {record['dt']} {record['Time']} - 收盘价: {record['cp']}")
    
    # 5. 测试批量操作
    logger.info("5. 测试批量操作...")
    batch_data = []
    for i in range(3):
        batch_data.append({
            'CodeID': 2,  # 对应AU2512的CodeID
            'dt': 20250929,
            'Time': 1430 + i,  # 14:30, 14:31, 14:32
            'MinuteValue': 870 + i,
            'hp': 260000 + i * 100,  # 假设金价
            'lp': 259000 + i * 100,
            'op': 259500 + i * 100,
            'cp': 260000 + i * 100,
            'zcp': 260000 + i * 100,
            'vol': 1000 + i,
            'amt': 520000 + i * 1000,
            'oi': 2000 + i,
            'rtn': 10 + i,
            'HQDataSourceType': '1'
        })
    
    success, error_code, error_msg, failed_records = hq_manager.batch_create(batch_data)
    if success:
        logger.info("批量创建行情记录成功")
    else:
        logger.error(f"批量创建行情记录失败 [{error_code}]: {error_msg}")
        if failed_records:
            logger.warning(f"失败记录数量: {len(failed_records)}")
            for i, record in enumerate(failed_records[:3]):  # 只显示前3条失败记录
                logger.warning(f"      失败记录{i+1}: CodeID={record.get('CodeID')}, dt={record.get('dt')}, Time={record.get('Time')}")
    
    # 测试重复键插入
    logger.info("测试重复键插入...")
    duplicate_data = {
        'CodeID': 1,  # 与之前的记录相同
        'dt': 20250929,
        'Time': 1430,
        'MinuteValue': 870,
        'hp': 377200,  # 不同的价格
        'lp': 376100,
        'op': 376600,
        'cp': 377200,
        'zcp': 377200,
        'vol': 15000,  # 不同的成交量
        'amt': 1100000000,
        'oi': 60000,
        'rtn': 75,
        'HQDataSourceType': '1'
    }
    
    success, error_code, error_msg = hq_manager.create(**duplicate_data)
    if success:
        logger.info(f"重复键处理成功 - 操作类型: {error_code}")
    else:
        logger.error(f"重复键处理失败 [{error_code}]: {error_msg}")
    
    # 6. 测试 FutureMinutesHQManager.create_or_update 函数
    logger.info("6. 测试 FutureMinutesHQManager.create_or_update 函数...")
    
    # 测试插入新记录
    logger.info("6.1 测试插入新记录...")
    new_record_data = {
        'hp': 378000,  # 最高价
        'lp': 377000,  # 最低价
        'op': 377500,  # 开盘价
        'cp': 378000,  # 收盘价
        'zcp': 378000,  # 结算价
        'vol': 20000,  # 成交量
        'amt': 1500000000,  # 成交额
        'oi': 70000,  # 持仓量
        'rtn': 100,  # 收益率
        'HQDataSourceType': '1'  # 数据源类型
    }
    
    success, operation_type, error_msg = hq_manager.create_or_update(
        CodeID=1,  # CU2511
        dt=20250929,
        Time=1435,  # 14:35
        **new_record_data
    )
    
    if success:
        logger.info(f"create_or_update 插入成功 - 操作类型: {operation_type}")
    else:
        logger.error(f"create_or_update 插入失败: {error_msg}")
    
    # 测试更新现有记录
    logger.info("6.2 测试更新现有记录...")
    update_record_data = {
        'hp': 379000,  # 更新最高价
        'lp': 378000,  # 更新最低价
        'cp': 379000,  # 更新收盘价
        'vol': 25000,  # 更新成交量
        'amt': 1800000000,  # 更新成交额
        'oi': 75000,  # 更新持仓量
        'rtn': 120,  # 更新收益率
    }
    
    success, operation_type, error_msg = hq_manager.create_or_update(
        CodeID=1,  # CU2511
        dt=20250929,
        Time=1435,  # 相同的时间点，应该触发更新
        **update_record_data
    )
    
    if success:
        logger.info(f"create_or_update 更新成功 - 操作类型: {operation_type}")
    else:
        logger.error(f"create_or_update 更新失败: {error_msg}")
    
    # 测试只提供部分字段的更新
    logger.info("6.3 测试部分字段更新...")
    partial_update_data = {
        'vol': 30000,  # 只更新成交量
        'oi': 80000,   # 只更新持仓量
    }
    
    success, operation_type, error_msg = hq_manager.create_or_update(
        CodeID=1,  # CU2511
        dt=20250929,
        Time=1435,  # 相同的时间点
        **partial_update_data
    )
    
    if success:
        logger.info(f"create_or_update 部分更新成功 - 操作类型: {operation_type}")
    else:
        logger.error(f"create_or_update 部分更新失败: {error_msg}")
    
    # 验证更新结果
    logger.info("6.4 验证更新结果...")
    updated_records = hq_manager.read(CodeID=1, dt=20250929, Time=1435)
    if updated_records:
        record = updated_records[0]
        logger.info(f"    更新后的记录: CodeID={record['CodeID']}, dt={record['dt']}, Time={record['Time']}")
        logger.info(f"    成交量: {record['vol']}, 持仓量: {record['oi']}")
        logger.info(f"    最高价: {record['hp']}, 收盘价: {record['cp']}")
    else:
        logger.error("未找到更新后的记录")
    
    # 测试错误情况 - 缺少必需字段
    logger.info("6.5 测试错误情况 - 缺少必需字段...")
    try:
        success, operation_type, error_msg = hq_manager.create_or_update(
            CodeID=999,  # 不存在的CodeID
            dt=20250929,
            Time=1440,
            vol=1000  # 只提供部分字段，缺少价格字段
        )
        
        if success:
            logger.info(f"处理缺少字段情况成功 - 操作类型: {operation_type}")
        else:
            logger.warning(f"处理缺少字段情况 - 操作类型: {operation_type}, 错误: {error_msg}")
    except Exception as e:
        logger.error(f"捕获异常: {str(e)}")
    
 
    
    success, error_code, error_msg = hq_manager.create(**duplicate_data)
    if success:
        logger.warning("重复键插入成功（不应该发生）")
    else:
        logger.info(f"重复键插入失败（预期结果） [{error_code}]: {error_msg}")
    
    # 6. 测试K线查询功能
    logger.info("6. 测试K线查询功能...")
    
    # 查询单个合约的K线数据
    logger.info("查询单个合约(CodeID=1)的K线数据:")
    df_single = hq_manager.query_klines(code_ids=1)
    logger.info(f"    查询结果: {len(df_single)} 条记录")
    if not df_single.empty:
        logger.info(f"    数据列: {list(df_single.columns)}")
        logger.info(f"    示例数据:")
        logger.info(f"      CodeID: {df_single.iloc[0]['CodeID']}, 日期: {df_single.iloc[0]['dt']}, 时间: {df_single.iloc[0]['Time']}, 收盘价: {df_single.iloc[0]['cp']}")
    
    # 查询多个合约的K线数据
    logger.info("查询多个合约(CodeID=[1,2])的K线数据:")
    df_multi = hq_manager.query_klines(code_ids=[1, 2])
    logger.info(f"    查询结果: {len(df_multi)} 条记录")
    if not df_multi.empty:
        unique_codes = df_multi['CodeID'].unique()
        logger.info(f"    包含合约: {list(unique_codes)}")
        for code_id in unique_codes:
            code_data = df_multi[df_multi['CodeID'] == code_id]
            logger.info(f"      CodeID {code_id}: {len(code_data)} 条记录")
    
    # 查询指定日期范围的K线数据
    logger.info("查询指定日期(20250929)的K线数据:")
    df_date = hq_manager.query_klines(code_ids=[1, 2], start_date=20250929, end_date=20250929)
    logger.info(f"    查询结果: {len(df_date)} 条记录")
    
    # 查询指定时间范围的K线数据
    logger.info("查询指定时间范围(1430-1432)的K线数据:")
    df_time = hq_manager.query_klines(code_ids=[1, 2], start_time=1430, end_time=1432)
    logger.info(f"    查询结果: {len(df_time)} 条记录")
    if not df_time.empty:
        logger.info(f"    时间范围: {df_time['Time'].min()} - {df_time['Time'].max()}")
    
    # 查询不存在的合约
    logger.info("查询不存在的合约(CodeID=999):")
    df_empty = hq_manager.query_klines(code_ids=999)
    logger.info(f"    查询结果: {len(df_empty)} 条记录")
    logger.info(f"    返回空DataFrame，列数: {len(df_empty.columns)}")
    
    # 7. 测试新增的 create_or_update 方法
    logger.info("7. 测试 create_or_update 方法...")
    
    # 测试 CodeTableManager 的 create_or_update 方法
    logger.info("测试 CodeTableManager.create_or_update:")
    
    # 插入新记录
    success, operation = code_manager.create_or_update(
        CodeID=10, Market=1, MarketDesc='期货', StockCode='AG2512', 
        StockName='沪银2512合约', ExchangeType=2, ExchangeTypeDesc='上海期货交易所'
    )
    if success:
        logger.info(f"    新记录操作成功: {operation}")
    else:
        logger.error(f"    新记录操作失败")
    
    # 更新现有记录
    success, operation = code_manager.create_or_update(
        CodeID=10, Market=1, MarketDesc='期货', StockCode='AG2512', 
        StockName='沪银2512合约', ExchangeType=2, ExchangeTypeDesc='上海期货交易所'
    )
    if success:
        logger.info(f"    更新记录操作成功: {operation}")
    else:
        logger.error(f"    更新记录操作失败")
    
    # 8. 测试CSV导入功能
    logger.info("8. 测试CSV导入功能...")